package com.jeffse.light.stream.app.function;

import com.jeffse.light.stream.app.pojo.User;
import com.jeffse.light.stream.core.data.context.Context;
import com.jeffse.light.stream.core.data.context.ProcessContext;
import com.jeffse.light.stream.core.data.descriptor.stateDescriptor.ListStateDescriptor;
import com.jeffse.light.stream.core.data.descriptor.stateDescriptor.MapStateDescriptor;
import com.jeffse.light.stream.core.data.descriptor.stateDescriptor.ValueStateDescriptor;
import com.jeffse.light.stream.core.data.state.ListState;
import com.jeffse.light.stream.core.data.state.MapState;
import com.jeffse.light.stream.core.data.state.ValueState;
import com.jeffse.light.stream.core.function.base.KeyedProcessFunction;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.List;

@Slf4j
public class ListFunction extends KeyedProcessFunction<String, User, User> {

        private ValueState<User> valueState;
        private ListState<User> listState;
        private MapState<String, User> mapState;

        @Override
        public void open() {
            super.open();
            valueState = getRuntimeContext().getState(new ValueStateDescriptor("valueState", this));
            listState = getRuntimeContext().getListState(new ListStateDescriptor("listState", this));
            mapState = getRuntimeContext().getMapState(new MapStateDescriptor("mapState", this));
        }

        @Override
        public void processElement(ProcessContext sourceFunction, User value) {
            List<User> userList = new ArrayList<>();
            if(listState.getList() != null){
                userList.addAll(listState.getList());
            }

            userList.add(value);
            listState.update(userList);
            log.error("------------------" + getCurrentKey());
            valueState.update(value);

            log.error("获取状态数据" + valueState.value().getName());
            log.error("获取状态数据列表" + userList.toString());
        }

}
